use std::{
    collections::BTreeSet,
    sync::{Arc, Mutex},
};

use anyhow::{Context, Result};
use iroh::{protocol::Router, PublicKey};
pub use iroh::{NodeId, SecretKey};
use iroh_base::{ticket::Ticket, Signature};
pub use iroh_gossip::proto::TopicId;
use iroh_gossip::{
    api::{Event as GossipEvent, GossipSender},
    net::{Gossip, GOSSIP_ALPN},
};
use n0_future::{
    boxed::BoxStream,
    task::{self, AbortOnDropHandle},
    time::{Duration, SystemTime},
    StreamExt,
};
use serde::{Deserialize, Serialize};
use tokio::sync::{Mutex as TokioMutex, Notify};
use tracing::{debug, info, warn};

pub const TOPIC_PREFIX: &str = "leetshell/0:";
pub const PRESENCE_INTERVAL: Duration = Duration::from_secs(5);

#[derive(Serialize, Deserialize, Clone, Debug)]
pub struct ChatTicket {
    pub topic_id: TopicId,
    pub bootstrap: BTreeSet<NodeId>,
}

impl ChatTicket {
    pub fn new_random() -> Self {
        let topic_id = TopicId::from_bytes(rand::random());
        Self::new(topic_id)
    }

    pub fn new(topic_id: TopicId) -> Self {
        Self {
            topic_id,
            bootstrap: Default::default(),
        }
    }
    pub fn deserialize(input: &str) -> Result<Self> {
        <Self as Ticket>::deserialize(input).map_err(Into::into)
    }
    pub fn serialize(&self) -> String {
        <Self as Ticket>::serialize(self)
    }
}

impl Ticket for ChatTicket {
    const KIND: &'static str = "riterm";

    fn to_bytes(&self) -> Vec<u8> {
        postcard::to_stdvec(&self).unwrap()
    }

    fn from_bytes(bytes: &[u8]) -> Result<Self, iroh_base::ticket::ParseError> {
        let ticket = postcard::from_bytes(bytes)?;
        Ok(ticket)
    }
}

pub struct ChatNode {
    secret_key: SecretKey,
    router: Router,
    gossip: Gossip,
}

impl ChatNode {
    /// Spawns a gossip node.
    pub async fn spawn(secret_key: Option<SecretKey>) -> Result<Self> {
        let secret_key = secret_key.unwrap_or_else(|| {
            let bytes = rand::random::<[u8; 32]>();
            SecretKey::from_bytes(&bytes)
        });
        let endpoint = iroh::Endpoint::builder()
            .secret_key(secret_key.clone())
            .discovery_n0()
            .alpns(vec![GOSSIP_ALPN.to_vec()])
            .bind()
            .await?;

        let node_id = endpoint.node_id();
        info!("endpoint bound");
        info!("node id: {node_id:#?}");

        let gossip = Gossip::builder().spawn(endpoint.clone());
        info!("gossip spawned");
        let router = Router::builder(endpoint)
            .accept(GOSSIP_ALPN, gossip.clone())
            .spawn();
        info!("router spawned");
        Ok(Self {
            gossip,
            router,
            secret_key,
        })
    }

    /// Returns the node id of this node.
    pub fn node_id(&self) -> NodeId {
        self.router.endpoint().node_id()
    }

    /// Returns information about all the remote nodes this [`Endpoint`] knows about.
    /// TODO: Implement with new IROH API
    pub fn remote_info(&self) -> Vec<()> {
        Vec::new()
    }

    /// Returns the endpoint for direct access.
    pub fn endpoint(&self) -> &iroh::Endpoint {
        self.router.endpoint()
    }

    /// Joins a chat channel from a ticket.
    ///
    /// Returns a [`ChatSender`] to send messages or change our nickname
    /// and a stream of [`Event`] items for incoming messages and other event.s
    pub async fn join(
        &self,
        ticket: &ChatTicket,
        nickname: String,
    ) -> Result<(ChatSender, BoxStream<Result<Event>>)> {
        let topic_id = ticket.topic_id;
        let bootstrap = ticket.bootstrap.iter().cloned().collect();
        info!(?bootstrap, "joining {topic_id}");
        let gossip_topic = self.gossip.subscribe(topic_id, bootstrap).await?;
        let (sender, receiver) = gossip_topic.split();

        let nickname = Arc::new(Mutex::new(nickname));
        let trigger_presence = Arc::new(Notify::new());

        // We spawn a task that occasionally sens a Presence message with our nickname.
        // This allows to track which peers are online currently.
        let sender = Arc::new(TokioMutex::new(sender));
        let presence_task = AbortOnDropHandle::new(task::spawn({
            let secret_key = self.secret_key.clone();
            let sender = sender.clone();
            let trigger_presence = trigger_presence.clone();
            let nickname = nickname.clone();

            async move {
                loop {
                    let nickname = nickname.lock().expect("poisened").clone();
                    let message = NetMessage::Chat {
                        text: String::new(),
                        nickname,
                        timestamp: 0,
                    };
                    debug!("send presence {message:?}");
                    let signed_message = SignedMessage::sign_and_encode(&secret_key, message)
                        .expect("failed to encode message");
                    if let Err(err) = sender.lock().await.broadcast(signed_message.into()).await {
                        tracing::warn!("presence task failed to broadcast: {err}");
                        break;
                    }
                    n0_future::future::race(
                        n0_future::time::sleep(PRESENCE_INTERVAL),
                        trigger_presence.notified(),
                    )
                    .await;
                }
            }
        }));

        // We create a stream of events, coming from the gossip topic event receiver.
        // We'll want to map the events to our own event type, which includes parsing
        // the messages and verifying the signatures, and trigger presence
        // once the swarm is joined initially.
        let receiver = n0_future::stream::try_unfold(receiver, {
            let trigger_presence = trigger_presence.clone();
            move |mut receiver| {
                let trigger_presence = trigger_presence.clone();
                async move {
                    loop {
                        // Store if we were joined before the next event comes in.
                        let was_joined = receiver.is_joined();

                        // Fetch the next event.
                        let Some(event) = receiver.try_next().await? else {
                            return Ok(None);
                        };
                        // Convert into our event type. this fails if we receive a message
                        // that cannot be decoced into our event type. If that is the case,
                        // we just keep and log the error.
                        let event: Event = match event.try_into() {
                            Ok(event) => event,
                            Err(err) => {
                                warn!("received invalid message: {err}");
                                continue;
                            }
                        };
                        // If we just joined, trigger sending our presence message.
                        if !was_joined && receiver.is_joined() {
                            trigger_presence.notify_waiters()
                        };

                        break Ok(Some((event, receiver)));
                    }
                }
            }
        });

        let sender = ChatSender {
            secret_key: self.secret_key.clone(),
            nickname,
            sender,
            trigger_presence,
            _presence_task: Arc::new(presence_task),
        };
        Ok((sender, Box::pin(receiver)))
    }

    pub async fn shutdown(&self) {
        if let Err(err) = self.router.shutdown().await {
            warn!("failed to shutdown router cleanly: {err}");
        }
        self.router.endpoint().close().await;
    }
}

#[derive(Debug, Clone)]
pub struct ChatSender {
    nickname: Arc<Mutex<String>>,
    secret_key: SecretKey,
    sender: Arc<TokioMutex<GossipSender>>,
    trigger_presence: Arc<Notify>,
    _presence_task: Arc<AbortOnDropHandle<()>>,
}

impl ChatSender {
    pub async fn send(&self, text: String) -> Result<()> {
        let nickname = self.nickname.lock().expect("poisened").clone();
        let message = NetMessage::Chat {
            text,
            nickname,
            timestamp: 0,
        };
        let signed_message = SignedMessage::sign_and_encode(&self.secret_key, message)?;
        self.sender
            .lock()
            .await
            .broadcast(signed_message.into())
            .await?;
        Ok(())
    }

    pub fn set_nickname(&self, name: String) {
        *self.nickname.lock().expect("poisened") = name;
        self.trigger_presence.notify_waiters();
    }

    // Optimized batch terminal operations
    pub async fn send_terminal_batch(&self, operations: Vec<TerminalOp>) -> Result<()> {
        let message = NetMessage::TerminalBatch {
            operations,
            timestamp: 0,
        };
        let signed_message = SignedMessage::sign_and_encode(&self.secret_key, message)?;
        self.sender
            .lock()
            .await
            .broadcast(signed_message.into())
            .await?;
        Ok(())
    }

    // Convenience methods for single terminal operations
    pub async fn create_terminal(
        &self,
        session_id: String,
        shell_path: Option<String>,
        name: Option<String>,
    ) -> Result<()> {
        let operation = TerminalOp::Create {
            session_id,
            shell_path,
            name,
        };
        self.send_terminal_batch(vec![operation]).await
    }

    pub async fn send_terminal_input(&self, session_id: String, data: Vec<u8>) -> Result<()> {
        let operation = TerminalOp::Input { session_id, data };
        self.send_terminal_batch(vec![operation]).await
    }

    pub async fn resize_terminal(&self, session_id: String, rows: u32, cols: u32) -> Result<()> {
        let operation = TerminalOp::Resize {
            session_id,
            rows,
            cols,
        };
        self.send_terminal_batch(vec![operation]).await
    }

    pub async fn close_terminal(&self, session_id: String) -> Result<()> {
        let operation = TerminalOp::Close { session_id };
        self.send_terminal_batch(vec![operation]).await
    }

    /// Rename a terminal session - optimized batch operation
    pub async fn rename_terminal(
        &self,
        session_id: String,
        new_name: Option<String>,
    ) -> Result<()> {
        let operation = TerminalOp::Rename {
            session_id,
            new_name,
        };
        self.send_terminal_batch(vec![operation]).await
    }

    pub async fn send_terminal_output(&self, session_id: String, data: Vec<u8>) -> Result<()> {
        let operation = TerminalOp::Output { session_id, data };
        self.send_terminal_batch(vec![operation]).await
    }

    pub async fn send_terminal_status(&self, session_id: String, status: String) -> Result<()> {
        let operation = TerminalOp::Status { session_id, status };
        self.send_terminal_batch(vec![operation]).await
    }

    // Optimized terminal state synchronization
    pub async fn send_terminal_state(&self, terminals: Vec<TerminalInfo>) -> Result<()> {
        let message = NetMessage::TerminalState {
            terminals,
            timestamp: 0,
        };
        let signed_message = SignedMessage::sign_and_encode(&self.secret_key, message)?;
        self.sender
            .lock()
            .await
            .broadcast(signed_message.into())
            .await?;
        Ok(())
    }

    // Simplified web sharing
    pub async fn send_web_share(&self, local_port: u16, public_port: u16) -> Result<()> {
        let message = NetMessage::WebShare {
            local_port,
            public_port,
            timestamp: 0,
        };
        let signed_message = SignedMessage::sign_and_encode(&self.secret_key, message)?;
        self.sender
            .lock()
            .await
            .broadcast(signed_message.into())
            .await?;
        Ok(())
    }

    pub async fn send_web_unshare(&self, public_port: u16) -> Result<()> {
        let message = NetMessage::WebShareStop {
            public_port,
            timestamp: 0,
        };
        let signed_message = SignedMessage::sign_and_encode(&self.secret_key, message)?;
        self.sender
            .lock()
            .await
            .broadcast(signed_message.into())
            .await?;
        Ok(())
    }

    // WebShare state synchronization
    pub async fn send_web_share_state(&self, proxies: Vec<WebShareProxyInfo>) -> Result<()> {
        let message = NetMessage::WebShareState {
            proxies,
            timestamp: 0,
        };
        let signed_message = SignedMessage::sign_and_encode(&self.secret_key, message)?;
        self.sender
            .lock()
            .await
            .broadcast(signed_message.into())
            .await?;
        Ok(())
    }

    // Optimized batch method for multiple terminal operations
    pub async fn send_terminal_operations_batch(&self, operations: Vec<TerminalOp>) -> Result<()> {
        // Split into batches to avoid oversized messages
        const BATCH_SIZE: usize = 10;

        for chunk in operations.chunks(BATCH_SIZE) {
            let batch = chunk.to_vec();
            self.send_terminal_batch(batch).await?;
        }

        Ok(())
    }
}

#[derive(Debug, Clone, Serialize, Deserialize)]
#[serde(tag = "type", rename_all = "camelCase")]
pub enum NetMessage {
    // Network events
    NodeJoin {
        node_id: NodeId,
        timestamp: u64,
    },
    NodeLeave {
        node_id: NodeId,
        timestamp: u64,
    },

    // Chat messages
    Chat {
        text: String,
        nickname: String,
        timestamp: u64,
    },

    // Terminal batch operations - combines multiple terminal operations
    TerminalBatch {
        operations: Vec<TerminalOp>,
        timestamp: u64,
    },

    // Terminal state synchronization - complete state dump
    TerminalState {
        terminals: Vec<TerminalInfo>,
        timestamp: u64,
    },

    // Web sharing
    WebShare {
        local_port: u16,
        public_port: u16,
        timestamp: u64,
    },
    WebShareStop {
        public_port: u16,
        timestamp: u64,
    },

    // WebShare state synchronization
    WebShareState {
        proxies: Vec<WebShareProxyInfo>,
        timestamp: u64,
    },

    System {
        timestamp: u64,
    },
}

#[derive(Debug, Clone, Serialize, Deserialize)]
#[serde(tag = "op", rename_all = "camelCase")]
pub enum TerminalOp {
    Create {
        session_id: String,
        shell_path: Option<String>,
        name: Option<String>,
    },
    Input {
        session_id: String,
        data: Vec<u8>,
    },
    Output {
        session_id: String,
        data: Vec<u8>,
    },
    Resize {
        session_id: String,
        rows: u32,
        cols: u32,
    },
    Status {
        session_id: String,
        status: String,
    },
    Close {
        session_id: String,
    },
    Rename {
        session_id: String,
        new_name: Option<String>,
    },
}

// Simplified event wrapper for external consumers
#[derive(Debug, Clone, Serialize, Deserialize)]
#[serde(tag = "type", rename_all = "camelCase")]
pub enum Event {
    #[serde(rename_all = "camelCase")]
    NodeJoined {
        node_id: NodeId,
    },
    #[serde(rename_all = "camelCase")]
    NodeLeft {
        node_id: NodeId,
    },
    #[serde(rename_all = "camelCase")]
    MessageReceived {
        from: NodeId,
        text: String,
        nickname: String,
    },
    #[serde(rename_all = "camelCase")]
    TerminalBatch {
        from: NodeId,
        operations: Vec<TerminalOp>,
    },
    #[serde(rename_all = "camelCase")]
    TerminalState {
        from: NodeId,
        terminals: Vec<TerminalInfo>,
    },
    #[serde(rename_all = "camelCase")]
    WebShare {
        from: NodeId,
        local_port: u16,
        public_port: u16,
    },
    #[serde(rename_all = "camelCase")]
    WebShareStop {
        from: NodeId,
        public_port: u16,
    },
    #[serde(rename_all = "camelCase")]
    WebShareState {
        from: NodeId,
        proxies: Vec<WebShareProxyInfo>,
    },
    Lagged,
}

impl TryFrom<GossipEvent> for Event {
    type Error = anyhow::Error;
    fn try_from(event: GossipEvent) -> Result<Self, Self::Error> {
        let converted = match event {
            GossipEvent::NeighborUp(node_id) => Self::NodeJoined { node_id },
            GossipEvent::NeighborDown(node_id) => Self::NodeLeft { node_id },
            GossipEvent::Received(message) => {
                let message = SignedMessage::verify_and_decode(&message.content)
                    .context("failed to parse and verify signed message")?;
                match message.message {
                    NetMessage::NodeJoin {
                        node_id,
                        timestamp: _,
                    } => Self::NodeJoined { node_id },
                    NetMessage::NodeLeave {
                        node_id,
                        timestamp: _,
                    } => Self::NodeLeft { node_id },
                    NetMessage::Chat {
                        text,
                        nickname,
                        timestamp: _,
                    } => {
                        if text.is_empty() {
                            // Empty chat messages are used as presence indicators
                            Self::MessageReceived {
                                from: message.from,
                                text: "".to_string(),
                                nickname,
                            }
                        } else {
                            Self::MessageReceived {
                                from: message.from,
                                text,
                                nickname,
                            }
                        }
                    }
                    NetMessage::TerminalBatch {
                        operations,
                        timestamp: _,
                    } => Self::TerminalBatch {
                        from: message.from,
                        operations,
                    },
                    NetMessage::TerminalState {
                        terminals,
                        timestamp: _,
                    } => Self::TerminalState {
                        from: message.from,
                        terminals,
                    },
                    NetMessage::WebShare {
                        local_port,
                        public_port,
                        timestamp: _,
                    } => Self::WebShare {
                        from: message.from,
                        local_port,
                        public_port,
                    },
                    NetMessage::WebShareStop {
                        public_port,
                        timestamp: _,
                    } => Self::WebShareStop {
                        from: message.from,
                        public_port,
                    },
                    NetMessage::WebShareState {
                        proxies,
                        timestamp: _,
                    } => Self::WebShareState {
                        from: message.from,
                        proxies,
                    },
                    NetMessage::System { timestamp: _ } => Self::Lagged,
                }
            }
            GossipEvent::Lagged => Self::Lagged,
        };
        Ok(converted)
    }
}

#[derive(Debug, Serialize, Deserialize)]
struct SignedMessage {
    from: PublicKey,
    data: Vec<u8>,
    signature: Signature,
}

impl SignedMessage {
    pub fn verify_and_decode(bytes: &[u8]) -> Result<ReceivedMessage> {
        let signed_message: Self = postcard::from_bytes(bytes)?;
        let key: PublicKey = signed_message.from;
        key.verify(&signed_message.data, &signed_message.signature)?;
        let message: WireMessage = postcard::from_bytes(&signed_message.data)?;
        let WireMessage::VO { timestamp, message } = message;
        Ok(ReceivedMessage {
            from: signed_message.from,
            timestamp,
            message,
        })
    }

    pub fn sign_and_encode(secret_key: &SecretKey, message: NetMessage) -> Result<Vec<u8>> {
        let timestamp = SystemTime::now()
            .duration_since(SystemTime::UNIX_EPOCH)
            .unwrap()
            .as_micros() as u64;
        let wire_message = WireMessage::VO { timestamp, message };
        let data = postcard::to_stdvec(&wire_message)?;
        let signature = secret_key.sign(&data);
        let from: PublicKey = secret_key.public();
        let signed_message = Self {
            from,
            data,
            signature,
        };
        let encoded = postcard::to_stdvec(&signed_message)?;
        Ok(encoded)
    }
}

#[derive(Debug, Serialize, Deserialize)]
pub enum WireMessage {
    VO { timestamp: u64, message: NetMessage },
}

#[derive(Debug, Serialize, Deserialize, Clone)]
pub struct TerminalInfo {
    pub session_id: String,
    pub name: Option<String>,
    pub position: TerminalPosition,
    pub size: TerminalSize,
    pub status: String,
    pub created_at: u64,
    pub last_activity: u64,
}

#[derive(Debug, Serialize, Deserialize, Clone)]
pub struct TerminalPosition {
    pub x: i32,
    pub y: i32,
}

#[derive(Debug, Serialize, Deserialize, Clone)]
pub struct TerminalSize {
    pub rows: u32,
    pub cols: u32,
}

#[derive(Debug, Serialize, Deserialize, Clone)]
pub struct WebShareProxyInfo {
    pub local_port: u16,
    pub public_port: u16,
    pub node_id: NodeId,
    pub status: String, // "active", "stopping", "error"
    pub created_at: u64,
}

#[derive(Debug, Serialize, Deserialize)]
pub struct ReceivedMessage {
    timestamp: u64,
    from: NodeId,
    message: NetMessage,
}
